package com.jorado.metrics.discovery;

import com.jorado.metrics.util.ApplicationHelp;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 需要增加 断开重连
 */
public class ZookeeperDiscovery implements AutoDiscovery {

    public static final Logger logger = LoggerFactory.getLogger(ZookeeperDiscovery.class);

    private CuratorFramework framework;
    private String zkString;
    private String path;
    private String value;

    private final Object monitorLock = new Object();

    private final ConnectionStateListener connectionStateListener = (client, state) -> {

        logger.info("监听服务zookeeper状态改变 {}", state);
        if (state == ConnectionState.RECONNECTED || state == ConnectionState.CONNECTED) {
            register();
        } else if (state == ConnectionState.LOST) {
            synchronized (this.monitorLock) {
                this.monitorLock.notifyAll();
            }
        }

    };

    public ZookeeperDiscovery(String zkString, String proxy, String host, int port, String context) {
        this.zkString = zkString;
        String discoverPath = "/metrics_discover";
        String format = "{\n" +
                "  \"ServiceEndpoint\":{\n" +
                "    \"Host\":\"%s/%s\",\n" +
                "    \"Port\":%d\n" +
                "  }\n" +
                "}";
        this.path = discoverPath + "/" + ApplicationHelp.getAppName() + "-" + host + ":" + port + "-" + UUID.randomUUID().toString();
        String metricsUrl = String.format("%s:%d", host, port);
        this.value = String.format(format, proxy, metricsUrl, port);
        this.start();
        //
        Thread guardThread = new Thread(() -> {
            try {
                // 开始后停顿1分钟
                Random random = new Random();
                int value = random.nextInt(30);
                Thread.sleep(value * 1000);
            } catch (Throwable cause) {
                //
            }
            while (true) {
                try {
                    deamonWatchMonitor();
                } catch (Throwable cause) {
                    //
                }
            }

        });
        guardThread.setDaemon(true);
        //
        guardThread.setName("watch-monitor-thread");
        guardThread.start();
    }

    private void deamonWatchMonitor() {
        synchronized (this.monitorLock) {
            try {
                this.monitorLock.wait(60 * 1000);
            } catch (Throwable cause) {
                //
            }
        }
        if (!isZookeeperConnected(getZkClient())) {
            logger.error("zookeeper修复连接{}", this.zkString);
            CuratorFramework zookeeperClient = null;
            boolean connectState = false;
            try {
                //
                zookeeperClient = CuratorFrameworkFactory.builder()
                        .connectString(this.zkString)
                        .retryPolicy(new ExponentialBackoffRetry(10000, 16))
                        .connectionTimeoutMs(3000)
                        .sessionTimeoutMs(30000)
                        .build();
                zookeeperClient.start();
                zookeeperClient.getConnectionStateListenable().addListener(this.connectionStateListener);
                zookeeperClient.blockUntilConnected(60, TimeUnit.SECONDS);
                if (isZookeeperConnected(zookeeperClient)) {
                    connectState = true;
                }
            } catch (Throwable cause) {
                connectState = false;
                slightCloseZkClient(zookeeperClient);
            }
            if (connectState) {
                logger.error("zookeeper修复连接{}, 结果: true", this.zkString);
                CuratorFramework oldZkClient = getZkClient();
                setZkClient(zookeeperClient);
                slightCloseZkClient(oldZkClient);
            } else {
                logger.error("zookeeper修复连接{}, 结果: false", this.zkString);
                slightCloseZkClient(zookeeperClient);
            }
        }
        //
        if (isZookeeperConnected(getZkClient())) {
            register();
        }
    }

    public void start() {
        logger.info("监控服务开启zookeeper: {}", this.zkString);
        try {
            this.framework = CuratorFrameworkFactory.builder()
                    .connectString(this.zkString)
                    .retryPolicy(new ExponentialBackoffRetry(10000, 5))
                    .connectionTimeoutMs(3000)
                    .sessionTimeoutMs(30000)
                    .build();
            this.framework.start();
            this.framework.getConnectionStateListenable().addListener(this.connectionStateListener);
        } catch (Exception cause) {
            logger.error("监控服务开启zookeeper: {} 异常", this.zkString, cause);
        }
    }

    @Override
    public void register() {
        try {
            if (this.framework.checkExists().forPath(this.path) == null) {
                String result = this.framework.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL).forPath(this.path, this.value.getBytes());
                logger.info("监控服务注册临时节点: {}, 结果: {}", this.path, result);
            }
        } catch (Exception cause) {
            logger.error("监控服务注册临时节点: {} 异常", this.path, cause);
        }
    }

    private boolean isZookeeperConnected(CuratorFramework framework) {
        if (framework != null && framework.getZookeeperClient() != null && framework.getZookeeperClient().isConnected()) {
            return true;
        } else {
            return false;
        }
    }

    private CuratorFramework getZkClient() {
        return this.framework;
    }

    private void setZkClient(CuratorFramework zookeeperClient) {
        this.framework = zookeeperClient;
    }

    private void slightCloseZkClient(final Closeable closeable) {
        if (closeable != null) {
            try {
                closeable.close();
            } catch (Throwable cause) {
                //
            }
        }
    }
}
